#include "Avalon/Network/MulticastSocket.hpp"
#include <boost/asio.hpp>
#include <boost/bind.hpp>
#include "Avalon/IO/Buffer.hpp"
#include "Avalon/Network/SocketThreadPool.hpp"

using namespace Avalon;
using namespace Avalon::IO;
using namespace Avalon::Network;
using namespace Avalon::Threading;
using namespace boost;
using namespace boost::asio;
using namespace boost::asio::ip;
using namespace boost::posix_time;

struct MulticastSocket::SocketDetails {
  boost::asio::ip::address_v4 m_group;
  boost::asio::ip::address_v4 m_interface;
  boost::asio::ip::udp::endpoint m_listenEndpoint;
  boost::asio::ip::udp::endpoint m_senderEndpoint;
  boost::scoped_ptr<boost::asio::ip::udp::socket> m_socket; 

  SocketDetails(const IpAddress& group,
    const IpAddress& interface)
      : m_group(ip::address_v4::from_string(group.first)),
        m_interface(ip::address_v4::from_string(interface.first)), 
        m_listenEndpoint(ip::udp::v4(), group.second) {}
};

MulticastSocket::MulticastSocket(const IpAddress& group,
    const IpAddress& interface)
    : m_socketDetails(new SocketDetails(group, interface)),
      m_isOpen(false),
      m_receiveBufferSize(0) {}

MulticastSocket::~MulticastSocket() {
  Close();
}

int MulticastSocket::GetReceiveBufferSize() const {
  return m_receiveBufferSize;
}

void MulticastSocket::SetReceiveBufferSize(int size) {
  m_receiveBufferSize = size;
}

void MulticastSocket::Open() {
  lock_guard<recursive_mutex> lock(m_mutex);
  if(m_isOpen) {
    return;
  }
  m_socketDetails->m_socket.reset(new ip::udp::socket(
    SocketThreadPool::GetInstance().GetService(),
    m_socketDetails->m_listenEndpoint.protocol()));
  m_socketDetails->m_socket->set_option(ip::udp::socket::reuse_address(true));
  m_socketDetails->m_socket->bind(m_socketDetails->m_listenEndpoint);
  if(m_receiveBufferSize != 0) {
    socket_base::receive_buffer_size bufferSize(m_receiveBufferSize);
    m_socketDetails->m_socket->set_option(bufferSize);
  }
  m_socketDetails->m_socket->set_option(ip::multicast::join_group(
    m_socketDetails->m_group, m_socketDetails->m_interface));
  m_isOpen = true;
}

void MulticastSocket::Close() {
  lock_guard<recursive_mutex> lock(m_mutex);
  if(!m_isOpen) {
    return;
  }
  m_socketDetails->m_socket->close();
  m_isOpen = false;
}

Async<DatagramPacket>::Ptr MulticastSocket::Receive(Buffer* buffer) {
  lock_guard<recursive_mutex> lock(m_mutex);
  AsyncOwner<DatagramPacket>::Ptr packet = AsyncOwner<DatagramPacket>::Create();
  m_socketDetails->m_socket->async_receive_from(
    boost::asio::buffer(buffer->GetMutableData(), buffer->GetSize()),
    m_socketDetails->m_senderEndpoint, bind(&MulticastSocket::HandleRead, this,
    packet, buffer, placeholders::error, placeholders::bytes_transferred));
  return packet;
}

Async<void>::Ptr MulticastSocket::Send(const DatagramPacket& packet) {
  lock_guard<recursive_mutex> lock(m_mutex);
  Buffer dataCopy = packet.GetData();
  ip::udp::endpoint destination(ip::address::from_string(
    packet.GetAddress().first), packet.GetAddress().second);
  AsyncOwner<void>::Ptr result = AsyncOwner<void>::Create();
  m_socketDetails->m_socket->async_send_to(
    buffer(dataCopy.GetData(), dataCopy.GetSize()), destination,
    boost::bind(&MulticastSocket::HandleWrite, this, result, dataCopy,
    asio::placeholders::error));
  return result;
}

void MulticastSocket::HandleRead(const AsyncOwner<DatagramPacket>::Ptr& packet,
    Buffer* buffer, const boost::system::error_code& error, size_t size) {
  lock_guard<recursive_mutex> lock(m_mutex);
  if(!error) {
    packet->SetResult(DatagramPacket(static_cast<int>(size), *buffer,
      make_pair(m_socketDetails->m_senderEndpoint.address().to_string(),
      m_socketDetails->m_senderEndpoint.port())));
  }
}

void MulticastSocket::HandleWrite(const AsyncOwner<void>::Ptr& result,
    const Buffer& buffer, const boost::system::error_code& error) {
  lock_guard<recursive_mutex> lock(m_mutex);
  result->SetComplete();
}
